package com.dagger.utils;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

import javax.websocket.Session;

import com.dagger.constant.MessageConstants;
import com.dagger.domain.resp.MsgResponse;
import com.dagger.threads.DelayConsumerThread;

public abstract class AbstractMsgUtil<T> {
	private static WebsocketUtil websocketUtil = new WebsocketUtil();
	private static SseEmitterUtil sseEmitterUtil = new SseEmitterUtil();

	private Map<String, CopyOnWriteArraySet<T>> realtimeMsgUsers = new HashMap<String, CopyOnWriteArraySet<T>>();
	private Map<String, CopyOnWriteArraySet<T>> delayMsgUsers = new HashMap<String, CopyOnWriteArraySet<T>>();
	private Map<String, DelayConsumerThread> delayConsumerThreadCache = new ConcurrentHashMap<>();

	/**
	 * 添加用户
	 * 
	 * @param user
	 */
	public abstract void _addUser(String msgKey, T user);

	/**
	 * 删除用户
	 * 
	 * @param user
	 */
	public abstract void _removeUser(T user);

	/**
	 * 发送消息
	 * 
	 * @param message
	 * @param user
	 * @return
	 */

	public abstract boolean sendMessage(T user, Object message);

	/**
	 * 发送延时消息
	 * 
	 * @param bussinessType
	 * @param bussinessId
	 * @param msg
	 */
	public final void _sendDelayMessage(String bussinessType, String bussinessId, Object msg) {
		String msgKey = MessageConstants.getMsgKey(bussinessType, bussinessId);
		CopyOnWriteArraySet<T> users = delayMsgUsers.get(msgKey);
		if (users != null) {
			users.forEach(user -> {
				sendMessage(user, MsgResponse.newDelayMessage(bussinessType, bussinessId, msg));
			});
		}
	}

	/**
	 * 发送实时消息
	 * 
	 * @param bussinessType
	 * @param bussinessId
	 * @param msg
	 */
	public final void _sendRealtimeMessage(String bussinessType, String bussinessId, Object msg) {
		String msgKey = MessageConstants.getMsgKey(bussinessType, bussinessId);
		CopyOnWriteArraySet<T> users = realtimeMsgUsers.get(msgKey);
		if (users != null) {
			users.forEach(user -> {
				sendMessage(user, MsgResponse.newRealtimeMessage(bussinessType, bussinessId, msg));
			});
		}
	}

	/**
	 * 是否需要消息
	 * 
	 * @param bussinessType
	 * @param bussinessId
	 * @return
	 */
	public boolean _needMsg(String bussinessType, String bussinessId) {
		String msgKey = MessageConstants.getMsgKey(bussinessType, bussinessId);
		return realtimeMsgUsers.containsKey(msgKey) || delayMsgUsers.containsKey(msgKey);
	}

	/**
	 * 获取所有的消息
	 * 
	 * @return
	 */
	public Set<String> _getAllMsgKey() {
		Set<String> keys = new HashSet<String>();
		if (!realtimeMsgUsers.isEmpty()) {
			keys.addAll(realtimeMsgUsers.keySet());
		}

		if (!delayMsgUsers.isEmpty()) {
			keys.addAll(delayMsgUsers.keySet());
		}
		return keys;
	}

	/**
	 * 添加用户
	 * 
	 * @param msgKey
	 * @param user
	 */
	public final void addDelayUser(String msgKey, T user) {
		CopyOnWriteArraySet<T> users = delayMsgUsers.get(msgKey);
		if (users == null) {
			users = new CopyOnWriteArraySet<>();
			delayMsgUsers.put(msgKey, users);
		}
		users.add(user);
		_addUser(msgKey, user);
	}

	public final void addRealtimeUser(String msgKey, T user) {
		CopyOnWriteArraySet<T> users = realtimeMsgUsers.get(msgKey);
		if (users == null) {
			users = new CopyOnWriteArraySet<>();
			realtimeMsgUsers.put(msgKey, users);
		}
		users.add(user);
		_addUser(msgKey, user);
	}

	/**
	 * 删除用户
	 * 
	 * @param user
	 */
	public final synchronized void removeUser(T user) {
		if (!realtimeMsgUsers.isEmpty()) {
			Object[] keys = realtimeMsgUsers.keySet().toArray();
			for (Object key : keys) {
				unSubcribeMsg(user, key);
			}
		}

		if (!delayMsgUsers.isEmpty()) {
			Object[] keys = delayMsgUsers.keySet().toArray();
			for (Object key : keys) {
				unSubcribeMsg(user, key);
			}
		}

		_removeUser(user);
	}

	public final synchronized void subcribeDelayMsg(T user, String bussinessType, String bussinessId) {
		if (LocalBeanUtils.isSomeOneEmpty(bussinessType, bussinessId) || user == null) {
			return;
		}
		String msgKey = MessageConstants.getMsgKey(bussinessType, bussinessId);
		addDelayUser(msgKey, user);
		DelayConsumerThread delayConsumerThread = delayConsumerThreadCache.get(msgKey);
		if (delayConsumerThread == null) {
			try {
				delayConsumerThread = new DelayConsumerThread(bussinessType, bussinessId);
				delayConsumerThreadCache.put(msgKey, delayConsumerThread);
				delayConsumerThread.start();
				sendMessage(user, MsgResponse.newSuccessMessage(bussinessType, bussinessId, "延时消息订阅成功"));
			} catch (Exception e) {
				if (user instanceof Session) {
					// ws
					sendMessage(user,
							MsgResponse.newFailedMessage(bussinessType, bussinessId, "延时消息订阅失败：" + e.getMessage()));
				} else {
					throw e;
				}
			}
		}
	}

	/**
	 * 订阅实时消息
	 * 
	 * @param sseEmitter
	 * @param bussinessType
	 * @param bussinessId
	 */
	public void subcribeRealtimeMsg(T user, String bussinessType, String bussinessId) {
		if (LocalBeanUtils.isSomeOneEmpty(bussinessType, bussinessId) || user == null) {
			return;
		}
		String msgKey = MessageConstants.getMsgKey(bussinessType, bussinessId);
		addRealtimeUser(msgKey, user);
		sendMessage(user, MsgResponse.newSuccessMessage(bussinessType, bussinessId, "实时消息订阅成功"));
	}

	public final synchronized void unSubcribeMsg(T user, String bussinessType, String bussinessId) {
		if (LocalBeanUtils.isSomeOneEmpty(bussinessType, bussinessId) || user == null) {
			return;
		}
		String msgKey = MessageConstants.getMsgKey(bussinessType, bussinessId);
		unSubcribeMsg(user, msgKey);
		sendMessage(user, MsgResponse.newSuccessMessage(bussinessType, bussinessId, "消息取消订阅成功"));
	}

	private synchronized void unSubcribeMsg(T user, Object msgKey) {
		CopyOnWriteArraySet<T> users = delayMsgUsers.get(msgKey);
		if (users != null) {
			users.remove(user);
			if (users.isEmpty()) {
				delayMsgUsers.remove(msgKey);
			}
		}

		users = realtimeMsgUsers.get(msgKey);
		if (users != null) {
			users.remove(user);
			if (users.isEmpty()) {
				realtimeMsgUsers.remove(msgKey);
			}
		}
	}

	/**
	 * 发送延时消息
	 * 
	 * @param bussinessType
	 * @param bussinessId
	 * @param msg
	 */
	public static void sendDelayMessage(String bussinessType, String bussinessId, Object msg) {
		websocketUtil._sendDelayMessage(bussinessType, bussinessId, msg);
		sseEmitterUtil._sendDelayMessage(bussinessType, bussinessId, msg);
	}

	public static void sendRealtimeMessage(String bussinessType, String bussinessId, Object msg) {
		websocketUtil._sendRealtimeMessage(bussinessType, bussinessId, msg);
		sseEmitterUtil._sendRealtimeMessage(bussinessType, bussinessId, msg);
	}

	/**
	 * 获取消息key
	 * 
	 * @param user
	 * @param bussinessType
	 * @param bussinessId
	 * @return
	 */

	public static String getMsgKey(Object user, String bussinessType, String bussinessId) {
		if (user instanceof String) {
			return String.format("SSE:%s:%s", bussinessType, bussinessId);
		}
		if (user instanceof Session) {
			return String.format("WS:%s:%s", bussinessType, bussinessId);
		}

		throw new RuntimeException("not support user type: {}" + user.getClass().getName());
	}

	public static Set<String> getAllMsgKey() {
		Set<String> msgKeys = new HashSet<>();
		msgKeys.addAll(websocketUtil._getAllMsgKey());
		msgKeys.addAll(sseEmitterUtil._getAllMsgKey());
		return msgKeys;
	}

	public static WebsocketUtil getWebsocketUtil() {
		return websocketUtil;
	}

	public static SseEmitterUtil getSseEmitterUtil() {
		return sseEmitterUtil;
	}

	public static boolean needMsg(String bussinessType, String bussinessId) {
		return websocketUtil._needMsg(bussinessType, bussinessId)
				|| sseEmitterUtil._needMsg(bussinessType, bussinessId);
	}

}
